/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.news.server.util;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import java.util.HashSet;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 *
 * @author zwlin
 */
public class SenderPool {

	RESendSessionBuilder builder;
	private final BlockingQueue<RESendSession> buf;
	private final HashSet<RESendSession> set;
	private final int max;

	public SenderPool(RESendSessionBuilder builder, int maxParrallel) {
		this.builder = builder;
		this.max = maxParrallel <= 0 ? Integer.MAX_VALUE : maxParrallel;
		if (maxParrallel == 0) {
			buf = new LinkedBlockingQueue<>();
		} else {
			buf = new ArrayBlockingQueue<>(maxParrallel);
		}
		set = new HashSet<>();
	}

	public SenderPool(RESendSessionBuilder builder) {
		this(builder, 0);
	}

	public SenderPool(String namesrv, String topic, Map<String, REFieldType> types, int maxParrallel) throws REConnectionException {
		REConnection conn = new REConnection(namesrv);
		RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(topic);
		for (Map.Entry<String, REFieldType> entry : types.entrySet()) {
			builder.addColumn(entry.getKey(), entry.getValue(), true);
		}
		this.builder = builder;
		this.max = maxParrallel <= 0 ? Integer.MAX_VALUE : maxParrallel;
		if (maxParrallel == 0) {
			buf = new LinkedBlockingQueue<>();
		} else {
			buf = new ArrayBlockingQueue<>(maxParrallel);
		}
		set = new HashSet<>();
	}

	public SenderPool(String namesrv, String topic, Map<String, REFieldType> types) throws REConnectionException {
		this(namesrv, topic, types, 0);
	}

	public RESendSession Borrow() throws RESessionException, REConnectionException {
		while (true) {
			try {
				return buf.remove();
			} catch (NoSuchElementException e) {
				if (set.size() > max) {
					try {
						return buf.take();
					} catch (InterruptedException ex) {
					}
				} else {
					synchronized (set) {
						if (set.size() > max) {
						} else {
							RESendSession session = build();
							session.start();
							set.add(session);
							return session;
						}
					}
				}
			}
		}
	}

	public RESendSession tryBorrow() throws RESessionException, REConnectionException {
		try {
			return buf.remove();
		} catch (NoSuchElementException e) {
			if (set.size() > max) {
				return null;
			} else {
				synchronized (set) {
					if (set.size() > max) {
						return null;
					} else {
						RESendSession session = build();
						set.add(session);
						return session;
					}
				}
			}
		}
	}

	public boolean Return(RESendSession sender) {
		return set.contains(sender) && buf.add(sender);
	}

	private RESendSession build() throws RESessionException, REConnectionException {
		return (RESendSession) builder.build();
	}

}
